// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: fengyang

#include <filesystem>
#include <memory>

#include <cxxopts.hpp>

#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/exchange/base/executor/sync_executor.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/strategy/strategy.h"
#include "coin2/exchange/base/feed_util/kline.h"

#include "appcoin2/support/feed/interval_feed_tool/ohlc.h"
#include "appcoin2/support/feed/fastfeed_util.h"

// proto
#include "coin2/exchange/feed.pb.h"

using namespace coin2::base::config;
using namespace coin2::app;
using namespace coin2::exchange::base::symbology;
using namespace coin2::exchange::feed;
using namespace coin2::exchange::interval_feed_enum;
using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::symbology::IProduct;
using coin2::feed::fastfeed::FastFeedWriter;
using ::fastfeed::proto::BookBuilderParameter;
using coin2::exchange::di::ConstantInjectionBuilder;
using ::interval_feed::OhlcCalculator;

// 1 kline, wrap OhlcCalculator
class KlineCalculator {
 public:
  KlineCalculator() {}
  explicit KlineCalculator(
      int resolution,
      int64_t open_timestamp,
      double pre_close_price = std::numeric_limits<double>::quiet_NaN())
      : resolution_(resolution),
        open_timestamp_(open_timestamp),
        next_open_(open_timestamp + (int64_t)resolution * 1'000'000'000),
        pre_close_(pre_close_price),
        volume_(0) {
  }

  void Update(double price, double qty) {
    volume_ += qty;
    turnover_ += price * qty;
    ohlc_.Update(price);
  }

  double Open() const {
    double open = ohlc_.Get(OhlcType::OPEN);
    if (!std::isnan(open)) return open;
    if (!std::isnan(pre_close_)) return pre_close_;
    return 0;
  }

  double High() const {
    double high = ohlc_.Get(OhlcType::HIGH);
    if (!std::isnan(high)) return high;
    if (!std::isnan(pre_close_)) return pre_close_;
    return 0;
  }

  double Low() const {
    double low = ohlc_.Get(OhlcType::LOW);
    if (!std::isnan(low)) return low;
    if (!std::isnan(pre_close_)) return pre_close_;
    return 0;
  }

  double Close() const {
    double close = ohlc_.Get(OhlcType::CLOSE);
    if (!std::isnan(close)) return close;
    if (!std::isnan(pre_close_)) return pre_close_;
    return 0;
  }

  double Volume() const {
    return volume_;
  }

  double Turnover() const {
    return turnover_;
  }

  int64_t OpenTimestamp() const {
    return open_timestamp_;
  }

  int64_t NextOpenTimestamp() const {
    return next_open_;
  }

  void Export(int64_t timestamp, Klines* klines) {
    klines->UpdateClear();
    Kline* kline = klines->AddKline();
    kline->interval = KlineInterval(resolution_);
    kline->open_timestamp = open_timestamp_;
    kline->open = this->Open();
    kline->close = this->Close();
    kline->high = this->High();
    kline->low = this->Low();
    // volume/turnover value is not right in some cases
    // doesn't export to prevent from missleading user
    // kline->volume = this->Volume();
    // kline->turnover = this->Turnover();
    kline->timestamp = timestamp;
  }

 private:
  static fastfeed::proto::KlineInterval KlineInterval(int resolution) {
    static std::map<int, fastfeed::proto::KlineInterval> interval_map = {
      {60, fastfeed::proto::KLINE_INTERVAL_1MIN},
      {3 * 60, fastfeed::proto::KLINE_INTERVAL_3MIN},
      {5 * 60, fastfeed::proto::KLINE_INTERVAL_5MIN},
      {15 * 60, fastfeed::proto::KLINE_INTERVAL_15MIN},
      {30 * 60, fastfeed::proto::KLINE_INTERVAL_30MIN},
      {60 * 60, fastfeed::proto::KLINE_INTERVAL_1HOUR},
      {24 * 60 * 60, fastfeed::proto::KLINE_INTERVAL_1DAY}
    };
    auto iter = interval_map.find(resolution);
    if (iter == interval_map.end()) {
      NOTREACHED() << "Not support interval " << resolution;
      return fastfeed::proto::KLINE_INTERVAL_INVALID;
    }

    return iter->second;
  }

  int resolution_;
  int64_t open_timestamp_;
  int64_t next_open_;
  double pre_close_;
  double volume_;
  double turnover_;
  OhlcCalculator ohlc_;
};

// usd for 1 symbol
class SymbolKline {
 public:
  explicit SymbolKline(
      const MarketExchangeApi& mea,
      const IProduct& product,
      const std::string& out_dir,
      const std::string& out_machine,
      const std::string& recipe,
      int worker_id,
      const std::vector<int>& resolution,
      int64_t start_ts,
      int64_t end_ts)
      : symbol_(product.absolute_norm()),
        product_(product.Clone()),
        machine_(out_machine),
        recipe_(recipe),
        end_ts_(end_ts),
        upd_(mea.String(), mea),
        fastfeed_converter_(out_dir) {
    for (auto& r : resolution) {
      CHECK_EQ(SEC_IN_DAY % r, 0) << "resolution must algin";
      resolution_kline_cal_[r] = std::make_unique<KlineCalculator>(r, start_ts);
    }

    // create fastfeed converter for kline store
    DriverConfig app;
    FeedSystemConfig e;
    e.set_recipe(recipe);
    e.add_archive_recipes(recipe);
    auto exchanges = app.mutable_feed()->mutable_exchanges();
    exchanges->clear();
    (*exchanges)[mea.String()] = e;
    app.mutable_feed()->mutable_common()->set_type(
        FeedSystemType::FEED_SYSTEM_TYPE_ARCHIVE);
    auto archive = app.mutable_feed()->mutable_common()->mutable_archive();
    archive->set_worker(worker_id);
    archive->set_archive_type(ArchiveConfig_ArchiveType::ArchiveConfig_ArchiveType_FASTFEED);
    archive->set_machine(machine_);
    int64_t save_start = end_ts - SEC_IN_DAY * 1'000'000'000LL;
    auto* interval = app.mutable_feed()->mutable_common()->mutable_archive()->
    mutable_interval();
    interval->set_start(YmdFromTimestamp(save_start));
    interval->set_end(YmdFromTimestamp(end_ts));

    coin2::app::Driver empty_driver(app, &fastfeed_converter_);
    fastfeed_converter_.Init(&empty_driver);
  }

  ~SymbolKline() {
    SaveLastKline();
  }

  void Update(const FeedUpdate& upd) {
    CHECK_EQ(upd.product().absolute_norm(), symbol_);
    for (auto& [resolution, kline] : resolution_kline_cal_) {
      while (upd.timestamp() >= kline->NextOpenTimestamp()) {
        SaveKline(resolution, last_upd_timestamp_);
        MoveToNext(resolution);
      }
      kline->Update(upd.trade().price, upd.trade().fill_qty);
    }

    last_upd_timestamp_ = upd.timestamp();
  }

 private:
  void SaveKline(int r, int64_t timestamp) {
    Klines& klines = resolution_klines_[r];
    resolution_kline_cal_[r]->Export(timestamp, &klines);
    upd_.SetRecipe(recipe_);
    upd_.SetTimestamp(timestamp);
    upd_.SetType(symbol_, *(product_.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(timestamp);
    upd_.SetKlines(&klines);
    fastfeed_converter_.onKlineFeed(upd_);
  }

  void MoveToNext(int r) {
    resolution_kline_cal_[r] = std::make_unique<KlineCalculator>(
            r,
            resolution_kline_cal_[r]->NextOpenTimestamp(),
            resolution_kline_cal_[r]->Close());
  }

  void SaveLastKline() {
    for (auto& [resolution, cal] : resolution_kline_cal_) {
      while (cal->OpenTimestamp() < end_ts_) {
        SaveKline(resolution, last_upd_timestamp_);
        MoveToNext(resolution);
      }
    }
  }

  std::string symbol_;
  std::unique_ptr<IProduct> product_;
  std::string machine_;
  std::string recipe_;
  int64_t end_ts_ = 0;
  int64_t last_upd_timestamp_ = 0;
  std::map<int, std::unique_ptr<KlineCalculator>> resolution_kline_cal_;
  std::map<int, Klines> resolution_klines_;
  FeedUpdate upd_;
  FastFeedConverter fastfeed_converter_;

  static constexpr int SEC_IN_DAY = 60 * 60 * 24;
};

class KlineFeedConverter : public coin2::strategy::IStrategy {
 public:
  explicit KlineFeedConverter(
      const std::string& out_dir,
      const std::string& out_machine,
      const std::string& out_recipe,
      int out_worker_id,
      const std::vector<int>& resolution)
      : out_dir_(out_dir),
        out_machine_(out_machine),
        recipe_(out_recipe),
        worker_id_(out_worker_id),
        resolution_(resolution) {
  }

  void Init(::coin2::app::Driver* driver) override {
    const auto& config = driver->config().feed();

    // Warmup period should be handled by parser but ignored by strategy.
    auto interval = FromIntervalConfig(config.common().archive().interval());
    start_from_ = interval.first;
    end_ts_ = interval.second;
    auto start_time = PtimeFromTimestamp(start_from_);
    CHECK_EQ(start_time.time_of_day().total_nanoseconds(), 0) <<
        "Converter must start from the begin of a day ";
  }

  void onTradeFeed(const FeedUpdate& upd) override {
    if (upd.timestamp() < start_from_) {
      return;
    }

    auto mea = upd.mea();
    auto symbol = upd.product().absolute_norm();
    auto mea_s = std::make_tuple(mea, symbol);
    if (klines_.count(mea_s) == 0) {
      klines_[mea_s] = std::make_unique<SymbolKline>(
          upd.mea(),
          upd.product(),
          out_dir_,
          out_machine_,
          recipe_,
          worker_id_,
          resolution_,
          start_from_,
          end_ts_);
    }

    klines_[mea_s]->Update(upd);
  }

 private:
  std::string out_dir_;
  std::string out_machine_;
  std::string recipe_;
  int worker_id_;
  std::vector<int> resolution_;
  int64_t start_from_;
  int64_t end_ts_;
  std::map<std::tuple<MarketExchangeApi, std::string>,
           std::unique_ptr<SymbolKline>> klines_;
};

int main(int argc, char* argv[]) {
  DriverConfig app;
  cxxopts::Options opt("Kline Feed Converter", "Generate Kline");
  AddDriverOptions(&opt);
  opt.add_options()
      ("date", "trading date",
          cxxopts::value<std::string>())
      ("start-date", "start trading date to load trade data",
          cxxopts::value<std::string>())
      ("out-dir", "kline feed out directory",
          cxxopts::value<std::string>()->default_value(
              "/remote/iosg/coin-2/buckets/feed.derived.feed_cache.coin2"))
      ("out-machine", "fastfeed output machine",
          cxxopts::value<std::string>()->default_value(
              "feed-05.ap-northeast-1.aws"))
      ("out-recipe", "kline output recipe",
          cxxopts::value<std::string>()->default_value(
              "kline_by_trade"))
      ("out-worker", "kline output worker id",
          cxxopts::value<int>()->default_value(
              "0"))
      ("resolution", "kline resolution",
          cxxopts::value<std::vector<int>>()->default_value(
              "300,900"));

  auto res = opt.parse(argc, argv);
  CHECK(ParseDriverOptions(res, &app)) << "Fail to parse driver";

  // port optoin date to dirver config
  if (res.count("date") != 0U) {
    auto* interval = app.mutable_feed()->mutable_common()->mutable_archive()->
        mutable_interval();
    interval->set_start(res["start-date"].as<std::string>());

    int64_t end_ts = TimestampFromString(res["date"].as<std::string>()) +
        24 * 60 * 60 * 1'000'000'000LL;
    interval->set_end(YmdFromTimestamp(end_ts));
  }

  KlineFeedConverter converter(res["out-dir"].as<std::string>(),
                               res["out-machine"].as<std::string>(),
                               res["out-recipe"].as<std::string>(),
                               res["out-worker"].as<int>(),
                               res["resolution"].as<std::vector<int>>());

  ConstantInjectionBuilder ci_builder = ConstantInjectionBuilder();
  coin2::app::Driver driver(app, &converter, ci_builder);
  driver.Run();

  return 0;
}
